package realtime.app.dim;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.ververica.cdc.connectors.mysql.source.MySqlSource;
import com.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.ververica.cdc.debezium.JsonDebeziumDeserializationSchema;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.hadoop.hbase.client.Connection;
import realtime.app.func.DimBroadcastFunction;
import realtime.app.func.DimSinkFunction;
import realtime.common.EduConfig;
import realtime.util.EnvUtil;
import realtime.util.HbaseUtil;
import realtime.util.KafkaUtil;

import java.util.Properties;

/**
 * @Create 2023/9/5 19:02
 */
public class DimSinkApp {
    public static void main(String[] args) {
        //TODO 1、创建运行环境
        StreamExecutionEnvironment env = EnvUtil.getSEE(4);

        //TODO 2、从kafka读取数据
        DataStreamSource<String> KafkaDS = env.fromSource(KafkaUtil.getKafkaConsumer("topic_db", "dim_sink"), WatermarkStrategy.noWatermarks(), "topic_db");

        //TODO 3、对数据进行etl jsonstr->jsonobj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = KafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSON.parseObject(value);
                    String type = jsonObject.getString("type");
                    if (!(type.equals("bootstrap-complete") || type.equals("bootstrap-start"))) {
                        out.collect(jsonObject);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        });

        //TODO 4、从flinkCDC读取数据
        Properties props = new Properties();
        props.setProperty("useSSL", "false");
        MySqlSource<String> mySqlSource = MySqlSource.<String>builder()
                .hostname("localhost")
                .port(13306)
                .databaseList("edu_config")
                .tableList("edu_config.table_process")
                .username("root")
                .password("000000")
                .jdbcProperties(props)
                .serverTimeZone("Asia/Shanghai")
                .deserializer(new JsonDebeziumDeserializationSchema())
                .startupOptions(StartupOptions.initial())
                .build();

        DataStreamSource<String> mysqlDSSource = env.fromSource(mySqlSource,
                WatermarkStrategy.noWatermarks(),
                "MysqlSource");

        //TODO 5、将读取到的信息转换为jsonObj
        SingleOutputStreamOperator<JSONObject> cdcDS = mysqlDSSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String value, Collector<JSONObject> out) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                String op = jsonObject.getString("op");
                JSONObject DimJsonObj = new JSONObject();
                if (op.equals("d")) {
                    DimJsonObj = jsonObject.getJSONObject("before");
                } else {
                    DimJsonObj = jsonObject.getJSONObject("after");
                }
                DimJsonObj.put("op", op);
                out.collect(DimJsonObj);
            }
        });

        //TODO 6、通过cdc配置信息创建hbase表
        cdcDS=cdcDS.process(new ProcessFunction<JSONObject, JSONObject>() {
            private Connection conn;

            @Override
            public void open(Configuration parameters) throws Exception {
                conn=HbaseUtil.getHbaseConnection();
            }

            @Override
            public void close() throws Exception {
                HbaseUtil.closeHbaseConnection(conn);
            }

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, JSONObject>.Context ctx, Collector<JSONObject> out) throws Exception {
                String op = jsonObject.getString("op");
                if (op.equals("d")){
                    HbaseUtil.dropTable(conn, EduConfig.HBASE_NAMESPACE,jsonObject.getString("sink_table"));
                }else if (op.equals("r")||op.equals("c")){
                    HbaseUtil.createTable(conn,EduConfig.HBASE_NAMESPACE,jsonObject.getString("sink_table"),"info");
                }else {
                    HbaseUtil.dropTable(conn, EduConfig.HBASE_NAMESPACE,jsonObject.getString("sink_table"));
                    HbaseUtil.createTable(conn,EduConfig.HBASE_NAMESPACE,jsonObject.getString("sink_table"),"info");
                }
                out.collect(jsonObject);
            }

        });


        //TODO 7、将flinkcdc创建为广播流,并与主流数据相交
        MapStateDescriptor stateDescriptor = new MapStateDescriptor<String,JSONObject>("cdcDS", String.class, JSONObject.class);
        BroadcastStream<JSONObject> broadcast = cdcDS.broadcast(stateDescriptor);

        BroadcastConnectedStream<JSONObject, JSONObject> connectDS = jsonObjDS.connect(broadcast);

        //TODO 8、对主流数据进行处理,过滤出维度数据
        SingleOutputStreamOperator<JSONObject> processDS = connectDS.process(new DimBroadcastFunction(stateDescriptor));

        //TODO 9、将数据写入到hbase中
        processDS.addSink(new DimSinkFunction());

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
